package com.atguigu.sparkcore.sharevariable

import org.apache.spark.util.AccumulatorV2

/**
 * description ：自定义累加类
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/1/10 6:20
 * modified By ：
 * version:    : 1.0
 */
class Acc extends AccumulatorV2[Int, Map[String, Int]] {

  // 累加器变量
  private var count = 0
  private var sum = 0

  // 是否零值
  override def isZero: Boolean = count == 0 && sum == 0

  // 拷贝方法，将累加器拷贝给其他excutor(拷贝的时候，要将当前的值全部复制)
  override def copy(): AccumulatorV2[Int, Map[String, Int]] = {
    val accNew = new Acc
    accNew.count = this.count
    accNew.sum = this.sum
    accNew
  }

  // 重置累加器
  override def reset(): Unit = {
    count = 0
    sum = 0
  }

  // 累加器计数，核心方法
  override def add(v: Int): Unit = {
    sum += v
    count += 1
  }

  // 合并累加器，执行value方法时，从其他excutor拉取数据，并合并数据
  override def merge(other: AccumulatorV2[Int, Map[String, Int]]): Unit = {
    other match {
      case other: Acc =>
        this.count += other.count
        this.sum += other.sum
      case _ => throw new UnsupportedOperationException(s"cannot merge ${this.getClass.getName} with ${other.getClass.getName}")
    }
  }

  // 获取累加器值
  override def value: Map[String, Int] = {
    Map("sum" -> this.sum, "count" -> this.count)
  }
}
